package kafkax

import (
	"context"
	"github.com/confluentinc/confluent-kafka-go/kafka"
	"github.com/go-kratos/kratos/v2/log"
	"go-kratos-demo/internal/conf"
	"os"
	"strings"
)

var (
	clientId, _ = os.Hostname()
)

type KafkaProducer struct {
	producer *kafka.Producer
	log      *log.Helper
}

func NewKafkaProducer(conf *conf.Data, logger log.Logger) *KafkaProducer {

	log_ := log.NewHelper(log.With(logger, "x_module", "data/NewKafkaProducer"))
	log_.Info("[boot] start NewKafkaProducer dbr.")

	cMap := kafka.ConfigMap{
		"bootstrap.servers": strings.Join(conf.Kafka.Addresses, ","),
		"client.id":         clientId, // todo
		"acks":              "all",
	}
	producer, err := kafka.NewProducer(&cMap)
	if err != nil {
		log_.Fatalf("[boot] failed opening connection to producer: %v", err)
	}

	kp := &KafkaProducer{
		producer: producer,
		log:      log_,
	}

	log_.Info("[boot] NewKafkaProducer success.")

	return kp
}

type ProduceExtParams struct {
	GoId int32
	// Mess  []byte
	// Topic string
}

func (p *KafkaProducer) Produce(ctx context.Context, topic string, mess []byte, ext ProduceExtParams) (err error) {

	// 组装消息
	msg := kafka.Message{
		TopicPartition: kafka.TopicPartition{
			Topic:     &topic,
			Partition: kafka.PartitionAny,
		},
		Value: mess,
	}

	// 发送消息
	deliveryChan := make(chan kafka.Event, 1)
	err = p.producer.Produce(&msg, deliveryChan)
	if err != nil {
		p.log.WithContext(ctx).Errorf("core:kafka:produce:p.producer.Produce:%v", err)
	}

	// 获取结果
	e := <-deliveryChan
	m := e.(*kafka.Message)

	// 结果验证
	if m.TopicPartition.Error != nil {
		p.log.WithContext(ctx).Errorf("core:kafka:produce:m.TopicPartition.Error:%v", m.TopicPartition.Error)
		return m.TopicPartition.Error
	} else {
		p.log.WithContext(ctx).Infof("【%d】message to topic %s [%d] at offset %v\\n", ext.GoId, *m.TopicPartition.Topic, m.TopicPartition.Partition, m.TopicPartition.Offset)
	}
	close(deliveryChan)

	return
}
